package com.shujia.wc;

import java.io.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Reduce {
    public static void main(String[] args) {

        ExecutorService executorService = Executors.newFixedThreadPool(4);

        /**
         * 相同后缀结尾的文件由同一个线程读取处理
         *
         */

        HashMap<Integer, ArrayList<File>> map = new HashMap<>();


        File pfile = new File("java/data/tmp4");

        File[] files = pfile.listFiles();
        for (File file : files) {
            Integer partition = Integer.parseInt(file.getName().split("_")[1]);

            /**
             * 如果分区在map中不存在  创建新的arrayList 将数据保存进行
             * 如果已经存在直接将数据保存进行
             *
             */

            ArrayList<File> files1 = map.get(partition);
            if (files1 == null) {
                ArrayList<File> files2 = new ArrayList<>();
                files2.add(file);
                map.put(partition, files2);
            } else {
                files1.add(file);
            }
        }

        //启动线程
        for (Map.Entry<Integer, ArrayList<File>> entry : map.entrySet()) {
            //当前分区
            Integer partition = entry.getKey();
            //当前分区所需要读取的所有文件
            ArrayList<File> files1 = entry.getValue();


            ReduceTask reduceTask = new ReduceTask(files1, partition);
            //启动线程
            executorService.submit(reduceTask);

        }

        executorService.shutdown();


    }
}

/**
 * 每个task 处理完多个文件
 */
class ReduceTask extends Thread {

    ArrayList<File> files;
    int partition;

    public ReduceTask(ArrayList<File> files, int partition) {
        this.files = files;
        this.partition = partition;
    }

    @Override
    public void run() {
        try {


            /**
             * 统计每个单词数量
             *
             */
            HashMap<String, Long> map = new HashMap<>();

            for (File file : files) {
                BufferedReader bufferedReader = new BufferedReader(new FileReader(file));
                String word;
                while ((word = bufferedReader.readLine()) != null) {
                    Long aLong = map.get(word);
                    if (aLong != null) {
                        map.put(word, aLong + 1);
                    } else {
                        map.put(word, 1L);
                    }
                }

                bufferedReader.close();
            }

            BufferedWriter bufferedWriter = new BufferedWriter(new FileWriter("java/data/tmp5/" + partition));


            for (Map.Entry<String, Long> entry : map.entrySet()) {
                String word = entry.getKey();
                Long count = entry.getValue();

                bufferedWriter.write(word + "," + count);
                bufferedWriter.newLine();

            }
            bufferedWriter.close();


        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
